package cn.colins.raft.node;

import cn.colins.raft.LogManage;
import cn.colins.raft.RaftNode;
import cn.colins.raft.enity.*;
import cn.colins.raft.enums.NodeStatusEnums;
import cn.colins.raft.config.ServerConfig;
import cn.colins.raft.exception.RaftRunException;
import cn.colins.raft.rpc.session.RpcSession;
import cn.colins.raft.rpc.session.RpcSessionFactory;
import cn.colins.raft.state.StateMachines;
import cn.colins.raft.utils.ThreadPoolUtils;
import cn.hutool.core.collection.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @Author czl
 * @Description candidate 节点操作
 * @Date 2023/7/14 17:53
 * @Param
 * @return
 **/
public class CandidateRaftNode implements RaftNode {

    private static final Logger log = LoggerFactory.getLogger(CandidateRaftNode.class);

    private final static NodeStatusEnums NODE_TYPE = NodeStatusEnums.CANDIDATE;


    @Override
    public ClientResponse clientRequestHandler(ClientRequest command, List<ServerConfig> serverConfigs) {
        // 直接拒绝
        ClientResponse build = ClientResponse.builder().code(401).build();
        build.setRequestId(command.getRequestId());
        return build;
    }

    @Override
    public AppendEntriesPreCommitResult logPreCommitHandler(AppendEntriesPreCommit appendEntriesPreCommit) {
        // 都已经打算起义了，突然收到了leader的投喂，怎么办。。
        // 符合条件也和follow一样的处理了
        AppendEntriesPreCommitResult build = AppendEntriesPreCommitResult.builder().build();
        build.setRequestId(appendEntriesPreCommit.getRequestId());
        LogManage logManage = RaftNodeInfo.getInstance().getLogManage();
        logManage.preCommitLog(appendEntriesPreCommit.getPreCommitLogId(),appendEntriesPreCommit.getEntries());
        build.setSuccess(true);
        return build;
    }

    @Override
    public void logCommitHandler(AppendEntriesCommit appendEntriesCommit) {
        // 都已经打算起义了，突然收到了leader的投喂，怎么办。。
        // 符合条件也和follow一样的处理了
        LogManage logManage = RaftNodeInfo.getInstance().getLogManage();
        logManage.commitLog(appendEntriesCommit.getCommitLogId(),appendEntriesCommit.getCommitLogIndex());
    }

    @Override
    public LogIndexPullResult sendLogPullRequest(List<Long> pullLogIndex) {
        throw new RaftRunException("candidate不需要发起log拉取请求");
    }

    @Override
    public LogIndexPullResult logPullRequestHandler(LogIndexPull logIndexPull) {
        throw new RaftRunException("candidate不需要处理log拉取请求");
    }


    @Override
    public boolean callVoteRequest(List<ServerConfig> serverConfigs) throws ExecutionException, InterruptedException {
        if (CollectionUtil.isEmpty(serverConfigs)) {
            StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);
            log.error("只有一个节点，还发起什么投票？");
            return false;
        }
        // candidate 会发起投票请求
        RaftNodeInfo instance = RaftNodeInfo.getInstance();

        // 投票过程中 可能又收到了心跳或者日志，状态已经变为follow
        if (!NODE_TYPE.equals(RaftNodeInfo.getInstance().getCurrentNodeStatus())) {
            return false;
        }
        log.debug(" {}: 哈哈哈，我发起了投票", RaftNodeInfo.getInstance().getSelf().toString());
        List<Future<RequestVoteResult>> taskList = new ArrayList<>(serverConfigs.size());

        // 加上自己的一票 需要 大于= n/2+1
        // 所以直接 >= n/2 就算通过了
        // 但是注意此时如果已经存在leader，日志数又不比当前leader大，所以leader还是leader 具有一票否决权
        for (ServerConfig serverConfig : serverConfigs) {
            Future<RequestVoteResult> voteResultFuture = ThreadPoolUtils.sendAsyncMsgPool.submit(() -> {
                // 构建投票
                RequestVoteRPC voteRPC = RequestVoteRPC.builder().candidateId(instance.getSelf().toString())
                        .term(instance.getCurrentTerm())  // 成为候选 的时候任期就+1了
                        .lastLogIndex(instance.getLastLogIndex()).build();
                RpcSession<RequestVoteResult, RequestVoteRPC> voteRPCRpcSession = RpcSessionFactory.<RequestVoteResult, RequestVoteRPC>openSession(serverConfig, voteRPC);
                return voteRPCRpcSession == null ? null : voteRPCRpcSession.syncSend(1000L);
            });
            taskList.add(voteResultFuture);
        }
        // 投票过程中 可能状态又已经变为follow
        if (!NODE_TYPE.equals(RaftNodeInfo.getInstance().getCurrentNodeStatus())) {
            return false;
        }

        return StateMachines.voteResultHandler(taskList, serverConfigs.size());

    }


    @Override
    public synchronized RequestVoteResult voteRequestHandler(RequestVoteRPC voteRPC) {
        // candidate 需要处理投票请求
        // 可能两个节点同时成为了候选人者
        RaftNodeInfo instance = RaftNodeInfo.getInstance();
        long currentTerm = instance.getCurrentTerm();
        RequestVoteResult voteResult = RequestVoteResult.builder().build();
        voteResult.setRequestId(voteRPC.getRequestId());
        // 1.任期比我大，我直接就同意 并且退位
        // 2.因为自己的票已经投了，就这一种情况
        if (voteRPC.getTerm() > currentTerm) {
            // 状态变更
            StateMachines.becomeFollow(voteRPC.getTerm(), null, voteRPC.getCandidateId());
            // 变为了follow 还需要开启心跳检测
            RaftNodeService.createElectionTask();

            voteResult.setTerm(currentTerm);
            voteResult.setVoteGranted(true);
            log.debug(" {}: 我刚刚变成了候选，我认可你的实力，我退位：{}", instance.getSelf().toString(), voteRPC.getCandidateId());
            return voteResult;
        }
        voteResult.setTerm(currentTerm);
        voteResult.setVoteGranted(false);
        log.debug(" {}: 我刚刚变成了候选，我不认可你的实力，我必须跟你抢一下：{}", instance.getSelf().toString(), voteRPC.getCandidateId());
        return voteResult;

    }

    @Override
    public boolean callHeartBeatRequest(List<ServerConfig> serverConfigs) {
        // candidate 不需要发送心跳
        throw new RaftRunException("candidate-发送心跳 状态出错啦");
    }

    @Override
    public HeartBeatResult heartBeatHandler(HeartBeatRequest heartBeatRequest) {
        log.warn("candidate-处理心跳 ！！！！！！！！！！！！！！！");
        // 可能延迟收到心跳，或者日志，所以变为follow
        // 亦或者已经有leader产生了
        // 默认返回即可，状态不做处理，因为投票会被leader否决
        HeartBeatResult build = HeartBeatResult.builder().code(200).build();
        build.setRequestId(heartBeatRequest.getRequestId());
        return build;

    }
}
